package com.example.hadoopdemo.hdfs;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.Properties;

/**
 * @author jianwang9@iflytek.com
 * @create 2018-03-30 15:30
 **/
public class HdfsTest {
    static FileSystem fileSystem;
    private static Configuration conf;

    static {
        conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://111.231.82.192:8020/");
        try {
            fileSystem = FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testHdfsCreateDir() throws IOException {
        Path path = new Path("/data");
        fileSystem.mkdirs(path, FsPermission.getDefault());
    }


    @Test
    public void testHdfsCreateFile() throws IOException {
        Path path = new Path("/data/hello.txt");
        FSDataOutputStream fos = fileSystem.create(path);
        fos.write("你好".getBytes());
        fos.close();
    }

    @Test
    public void testWritable() throws IOException {
        Path path = new Path("/data/hello.seq");
        SequenceFile.Writer writer = SequenceFile.createWriter(fileSystem, conf, path, IntWritable.class, Text.class);
        writer.append(new IntWritable(1),new Text("2222"));
        writer.append(new IntWritable(1),new Text("2222"));
        writer.append(new IntWritable(1),new Text("2222"));
        writer.append(new IntWritable(1),new Text("2222"));
        writer.hflush();
        writer.close();
    }


    @Test
    public void testReadWritable() throws IOException {
        Path path = new Path("/data/hello.seq");
        SequenceFile.Reader reader = new SequenceFile.Reader(fileSystem, path, conf);
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        //sequencefile 每放一个键值对,后会置一个同步点  如果同步点位置不对的话.从sequence中获取值将会报错
        long position = reader.getPosition();

        while (reader.next(key,value)){
            String sync = reader.syncSeen() ? "*" :"";
            System.out.printf("[%s%s]\t%s\t%s\n",position,sync,key,value);
            position = reader.getPosition();
        }

        reader.close();
    }
}
